package com.csf.event;

import com.alibaba.fastjson.JSONObject;
import com.csf.entity.DiscussPost;
import com.csf.entity.Event;
import com.csf.entity.Message;
import com.csf.service.DiscussPostService;
import com.csf.service.ElasticSearchService;
import com.csf.service.MessageService;
import com.csf.util.CommunityActivation;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Component
public class EventConsumer implements CommunityActivation {

//    监理日志
    private static final Logger logger = LoggerFactory.getLogger(EventConsumer.class);

    @Autowired
    private MessageService messageService;

    @Autowired
    private DiscussPostService discussPostService;

    @Autowired
    private ElasticSearchService elasticSearchService;

    @KafkaListener(topics = {TOPIC_COMMENT,TOPIC_LIKE})
    public void handleEvent(ConsumerRecord record){
          if (record==null || record.value() == null){
              logger.error("消息内容为空");
              return;
          }
//        对事件进行解析
          Event event = JSONObject.parseObject(record.value().toString(),Event.class);
          if (event==null){
              logger.error("格式不正确");
              return;
          }
//        将解析的事件存入消息实体中
          Message message = new Message();
          message.setFromId(SYSTEM_COMMENT);
          message.setToId(event.getEntityUserId());
          message.setConversationId(event.getTopic());
          message.setCreateTime(new Date());

        Map<String,Object> map = new HashMap<>();
//        content的里面要添加的是通知界面的字段拼接，比如"某某评论了您的帖子，点击查看"，这些关键字需要在content中获取。
        map.put("entityType",event.getEntityType());
        map.put("entityId",event.getEntityId());
        map.put("userId",event.getUserId());

        if (event.getData() != null){
             for (Map.Entry<String,Object> entry:event.getData().entrySet()){
                  map.put(entry.getKey(), entry.getValue());
             }
        }
        String s = JSONObject.toJSONString(map);
        message.setContent(s);
        messageService.insertMessage(message);

    }

//    利用卡夫卡将新添加的帖子放入es中
    @KafkaListener(topics = {TOPIC_DISES})
    public void handleDisEs(ConsumerRecord record){
        if (record ==null && record.value() == null){
            logger.error("消息内容为空");
        }
        //        对事件进行解析
        Event event = JSONObject.parseObject(record.value().toString(),Event.class);
        if (event==null){
            logger.error("格式不正确");
            return;
        }
        DiscussPost discussPost = discussPostService.selectDisById(event.getEntityId());
        elasticSearchService.saveDiscussPostEs(discussPost);
    }

    @KafkaListener(topics = {TOPIC_DELETE})
    public void handleDelete(ConsumerRecord record){
        if (record ==null && record.value() == null){
            logger.error("消息内容为空");
        }
        //        对事件进行解析
        Event event = JSONObject.parseObject(record.value().toString(),Event.class);
        if (event==null){
            logger.error("格式不正确");
            return;
        }
        elasticSearchService.deleteDiscussPostEs(discussPostService.selectDisById(event.getEntityId()));
    }
}
